package demo.service;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import demo.config.SpringApplicationContextAware;

/**
 * 因子计算分发器：分发任务，并行计算，汇总结果
 * 
 * @author hanjy
 *
 */
@Service
public class FeatureComputeService {

    private static Logger logger = LoggerFactory.getLogger(FeatureComputeService.class);

    @Autowired
    private SpringApplicationContextAware springApplicationContextAware;

    private Map<String, Calculator> calculators;

    @PostConstruct
    private void init() {
        calculators = springApplicationContextAware.getBeansOfType(Calculator.class);
    }

    /**
     * 并行计算
     * 
     * @param userId
     * @param fields
     * @return
     */
    public Map<String, Object> parallelExecute(long userId, Map<String, Map<String, Object>> fields) {

        Map<Future<Map<String, Object>>, String> taskMap = new HashMap<>(calculators.size());

        for (Entry<String, Calculator> entry : calculators.entrySet()) {
            Calculator calculator = entry.getValue();
            Set<String> intersection = new HashSet<>(calculator.supportFeatures());
            intersection.retainAll(fields.keySet());
            if (intersection.isEmpty()) {
                continue;
            }
            Map<String, Map<String, Object>> someFields = fields.entrySet().stream()
                    .filter(e -> intersection.contains(e.getKey()))
                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

            taskMap.put(calculator.asyncCompute(userId, someFields), entry.getKey());
        }

        Map<String, Object> result = new HashMap<>(fields.size());

        for (Entry<Future<Map<String, Object>>, String> entry : taskMap.entrySet()) {
            Future<Map<String, Object>> future = entry.getKey();
            String calculatorName = entry.getValue();
            try {
                Map<String, Object> futureResult = future.get();
                result.putAll(futureResult);
            } catch (InterruptedException e) {
                logger.error("calculator {} InterruptedException,userId:{}", calculatorName, userId);
                Thread.currentThread().interrupt();
            } catch (ExecutionException e) {
                logger.error("calculator {} ExecutionException,userId:{}", calculatorName, userId);
            }
        }
        return result;
    }
}
